#pragma once
#include <iostream>
#include <functional>
#include <unordered_map>
#include <vector>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <thread>
#include <mutex>
#include <sys/timerfd.h>
#include <cassert>
#include <cstring>
#include <condition_variable>
using namespace std;
#define EPOLL_EVENT_MAX_SIZE 1024
#define BUFFER_DEFALUT_SIZE 1024
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF
#define LOG(level, format, ...)                                                                                        \
    do                                                                                                                 \
    {                                                                                                                  \
        if (level < INF)                                                                                               \
            break;                                                                                                     \
        time_t t = time(NULL);                                                                                         \
        struct tm *ltm = localtime(&t);                                                                                \
        char tmp[32] = {0};                                                                                            \
        strftime(tmp, 31, "%H:%M:%S", ltm);                                                                            \
        fprintf(stdout, "[%p %s %s:%d] " format "\n", (void *)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)
#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)
class Buffer
{
private:
    uint64_t _writeIdx;   // 写入位置
    uint64_t _readIdx;    // 读取位置
    vector<char> _buffer; // 读写缓冲区

public:
    // 获取读地址
    char *getWriteAddr() { return &_buffer[0] + _writeIdx; }
    // 获取写地址
    char *getReadAddr() { return &_buffer[0] + _readIdx; }
    // 获取头部空余空间大小
    uint64_t getHeadSpace() { return _readIdx; }
    // 获取尾部空余空间大小
    uint64_t getTailSpace() { return _buffer.size() - _writeIdx; }
    // 获取空余空间大小
    uint64_t getFreeSpace() { return getHeadSpace() + getTailSpace(); }
    // 获取可读空间大小
    uint64_t getReadSpace() { return _writeIdx - _readIdx; }
    // 确保剩余空间大小
    void ensureSpace(uint64_t len)
    {
        if (len <= getTailSpace())
            return;
        else if (len < getFreeSpace())
        {
            uint64_t tmp = getReadSpace();
            copy(&_buffer[0] + _readIdx, &_buffer[0] + _writeIdx, &_buffer[0]);
            _readIdx = 0;
            _writeIdx = tmp;
        }
        else
        {
            _buffer.resize(_writeIdx + len, 0);
        }
    }
    // 移动读取位置
    void moveReadPosition(uint64_t len)
    {
        assert(len <= getReadSpace());
        _readIdx += len;
    }
    // 移动写入位置
    void moveWritePosition(uint64_t len)
    {
        assert(len <= getTailSpace());
        _writeIdx += len;
    }
    // 向缓冲区写入数据
    void write(const void *data, uint64_t len)
    {

        ensureSpace(len);
        char *d = (char *)data;
        copy(d, d + len, getWriteAddr());
    }
    // 从缓冲区读取数据
    void read(const void *data, uint64_t len)
    {
        assert(len <= getReadSpace());
        copy(getReadAddr(), getReadAddr() + len, (char *)data);
    }
    // 获取 \n 位置
    char *FindCRLF()
    {
        char *res = (char *)memchr((void *)getReadAddr(), '\n', getReadSpace());
        return res;
    }

    Buffer() : _writeIdx(0), _readIdx(0), _buffer(BUFFER_DEFALUT_SIZE)
    {
    }

    Buffer(const Buffer &other) : _writeIdx(other._writeIdx), _readIdx(other._readIdx), _buffer(other._buffer) {}
    Buffer(const string &str) : _writeIdx(str.size()), _readIdx(0)
    {
        _buffer.resize(str.size());
        copy(&str[0], &str[0] + str.size(), &_buffer[0]);
    }
    Buffer(const void *data, uint64_t len)
    {
        const char *d = (char *)data;
        ensureSpace(len);
        copy(d, d + len, &_buffer[0]);
    }
    // 清空缓冲区
    void Clean()
    {
        _buffer.clear();
        _readIdx = _writeIdx = 0;
    }

public:
    // 写入缓冲区
    void Write(const void *data, uint64_t len)
    {
        write(data, len);
    }
    // 写入缓冲区(void*)并移动写入位置
    void WriteAndPush(const void *data, uint64_t len)
    {
        write(data, len);
        moveWritePosition(len);
    }
    // 写入缓冲区(Buffer)
    void WriteBuffer(Buffer &other)
    {
        write((void *)other.getReadAddr(), other.getReadSpace());
    }
    // 写入缓冲区(Buffer)并移动写入位置
    void WriteBufferAndPush(Buffer &other)
    {
        write((void *)other.getReadAddr(), other.getReadSpace());
        moveWritePosition(other.getReadSpace());
    }
    // 写入缓冲区(string)
    void WriteString(const string &str)
    {
        write((char *)&str[0], str.size());
    }
    // 写入缓冲区(string)并移动写入位置
    void WriteStringAndPush(const string &str)
    {
        write((char *)&str[0], str.size());
        moveWritePosition(str.size());
    }
    // 读取缓冲区(void*)
    void Read(const void *data, uint64_t len)
    {
        read(data, len);
    }
    // 读取缓冲区(void*)并移动读取位置
    void ReadAndPop(const void *data, uint64_t len)
    {
        read(data, len);
        moveReadPosition(len);
    }
    // 读取缓冲区(string)
    string ReadAsString(uint64_t len)
    {
        string s;
        s.resize(len);
        read((void *)&s[0], len);
        return s;
    }
    // 读取缓冲区(string)并移动读取位置
    string ReadAsStringAndPop(uint64_t len)
    {
        string s;
        s.resize(len);
        s = ReadAsString(len);
        moveReadPosition(len);
        return s;
    }
    // 按行(\n)读取数据(string)
    string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
            return "";
        return ReadAsStringAndPop(pos - getReadAddr() + 1);
    }
};

class Socket
{
private:
    int _socket;

private:
    // 创建 socket 描述符
    bool Create()
    {
        _socket = socket(AF_INET, SOCK_STREAM, 0);
        if (_socket < 0)
        {
            ERR_LOG("Create socket failed!");
            return false;
        }
        return true;
    }
    // 绑定 socket 地址
    bool Bind(const string &ip, uint16_t port)
    {
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        if (bind(_socket, (sockaddr *)&addr, len) == -1)
        {
            ERR_LOG("Bind socket failed");
            return false;
        }
        return true;
    }
    // 设置 socket 描述符监听
    bool Listen(int backlog = 1024)
    {
        if (listen(_socket, backlog) == -1)
        {
            ERR_LOG("Listen socket failed");
            return false;
        }
        return true;
    }
    // 设置端口复用
    void setReuseAddress()
    {
        int opt = 1;
        setsockopt(_socket, SOL_SOCKET, SO_REUSEADDR, (void *)&opt, sizeof(opt));
        opt = 1;
        setsockopt(_socket, SOL_SOCKET, SO_REUSEPORT, (void *)&opt, sizeof(opt));
    }
    // 设置 socket 描述符非阻塞
    void setNonBlock()
    {
        int flag = fcntl(_socket, F_GETFL, 0);
        fcntl(_socket, F_SETFL, flag | O_NONBLOCK);
    }

public:
    Socket(int fd = -1) : _socket(fd) {}
    ~Socket()
    {
        if (_socket != -1)
            close(_socket);
    }
    // 获取 socket 描述符
    int Fd() { return _socket; }
    // 获取新建连接 socket 描述符
    int Appect()
    {
        sockaddr_in addr;
        socklen_t len = sizeof(addr);
        int newFd = accept(_socket, (sockaddr *)&addr, &len);
        if (newFd == -1)
        {
            ERR_LOG("Accept socket failed");
            return -1;
        }
        return newFd;
    }
    // 新建连接
    bool Connect(const string &ip, uint16_t port)
    {
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        if (connect(_socket, (sockaddr *)&addr, len) == -1)
        {
            ERR_LOG("Connect socket failed");
            return false;
        }
        return true;
    }
    // 从 socket 描述符读取数据
    ssize_t Recv(void *buffer, ssize_t len, int flag = 0)
    {
        ssize_t size = recv(_socket, buffer, len, flag);
        if (size <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return 0;
            ERR_LOG("Recv message failed");
            return -1;
        }
        return size;
    }
    // 向 socket 描述符写入数据
    ssize_t Send(void *buffer, ssize_t len, int flag = 0)
    {
        ssize_t size = send(_socket, buffer, len, flag);
        if (size <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return 0;
            ERR_LOG("Send message failed");
            return -1;
        }
        return size;
    }
    // 从 socket 描述符非阻塞读取数据
    ssize_t NoBolckRecv(void *buffer, ssize_t len)
    {
        if (len <= 0)
            return 0;
        return Recv(buffer, len, MSG_DONTWAIT);
    }
    // 从 socket 描述符非阻塞写入数据
    size_t NoBolckSend(void *buffer, ssize_t len)
    {
        if (len <= 0)
            return 0;
        return Send(buffer, len, MSG_DONTWAIT);
    }
    // 关闭 socket 描述符
    void Close()
    {
        if (_socket != -1)
            close(_socket);
    }
    // 创建服务端
    bool CreateServer(uint16_t port, const string &ip = "0.0.0.0", bool flag = false)
    {
        if (Create() == false)
            return false;
        if (flag == true)
            setNonBlock();
        setReuseAddress();
        if (Bind(ip, port) == false)
            return false;
        if (Listen() == false)
            return false;
        return true;
    }
    // 创建客户端
    bool CreateClient(uint16_t port, const string &ip = "127.0.0.1")
    {
        if (Create() == false)
            return false;
        if (Connect(ip, port) == false)
            return false;
        return true;
    }
};
class Poller;
class EventLoop;
class Channel
{
private:
    using EventCallBack = function<void()>;
    int _socket;            // socket描述符
    EventLoop *_loop;       // 所属哪个EventLoop
    uint32_t _events;       // 监听事件
    uint32_t _revents;      // 就绪事件
    EventCallBack _readCb;  // 读事件回调
    EventCallBack _writeCb; // 写事件回调
    EventCallBack _erronCb; // 错误事件回调
    EventCallBack _closeCb; // 关闭连接回调
    EventCallBack _eventCb; // 任意事件回调

public:
    Channel(int socket, EventLoop *loop) : _socket(socket), _loop(loop), _events(0), _revents(0) {}
    int Fd() { return _socket; }
    ~Channel()
    {
        if (_socket != -1)
            close(_socket);
    }
    uint32_t GetEvent() { return _events; }                              // 获取监听事件
    void SetREvent(uint32_t revents) { _revents = revents; }             // 设置就绪事件
    void SetReadCallBack(EventCallBack readCb) { _readCb = readCb; }     // 设置读事件回调
    void SetWriteCallBack(EventCallBack writeCb) { _writeCb = writeCb; } // 设置写事件回调
    void SetErrnoCallBack(EventCallBack errnoCb) { _erronCb = errnoCb; } // 设置错误事件回调
    void SetCloseCallBack(EventCallBack closeCb) { _closeCb = closeCb; } // 设置关闭连接回调
    void SetEventCallBack(EventCallBack eventCb) { _eventCb = eventCb; } // 设置任意事件回调
    bool ReadAble() { return _events & EPOLLIN; }                        // 是否监听读事件
    bool WriteAble() { return _events & EPOLLOUT; }                      // 是否监听写事件
    // 设置读事件监听并挂入Epoll中
    void EnableRead()
    {
        _events |= EPOLLIN;
        Updata();
    }
    // 设置写事件监听并挂入Epoll中
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Updata();
    }
    // 取消读事件监听
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Updata();
    }
    // 取消写事件监听
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Updata();
    }
    // 取消所有事件监听
    void DisableAllEvent()
    {
        _events = 0;
        Updata();
    }
    // 从Epoll中移除事件
    void Remove();
    // 更新Epoll中监听事件
    void Updata();
    // 处理监听事件
    void HandleEvent()
    {
        if ((_revents & EPOLLIN) || (_revents & EPOLLPRI) || (_revents & EPOLLRDHUP))
        {

            if (_readCb)
                _readCb();
        }
        if (_revents & EPOLLOUT)
        {
            if (_writeCb)
                _writeCb();
        }
        else if (_revents & EPOLLERR)
        {
            if (_erronCb)
                _erronCb();
        }
        else if (_revents & EPOLLERR)
        {
            if (_closeCb)
                _closeCb();
        }
        if (_eventCb)
            _eventCb();
    }
};

class Poller
{
private:
    int _epfd;                                        // Epoll模型描述符
    struct epoll_event _events[EPOLL_EVENT_MAX_SIZE]; // 接收就绪事件数组
    unordered_map<int, Channel *> _channels;          // socket描述符和对应Channel的映射

private:
    // 更新Epoll监听事件
    void Updata(Channel *channel, int op)
    {
        int fd = channel->Fd();
        epoll_event event;
        event.data.fd = fd;
        event.events = channel->GetEvent();
        int ret = epoll_ctl(_epfd, op, fd, &event);
        if (ret == -1)
        {
            ERR_LOG("Epoll ctl failed");
        }
    }
    // 检测Epoll是否监听了某socket描述符
    bool HasChannel(Channel *channel)
    {
        int fd = channel->Fd();
        auto it = _channels.find(fd);
        if (it == _channels.end())
            return false;
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(EPOLL_EVENT_MAX_SIZE);
        if (_epfd == -1)
        {
            ERR_LOG("Epoll create failed");
            abort();
        }
    }
    // 插入或修改Epoll的监听socket
    void UpdataEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == true)
        {
            Updata(channel, EPOLL_CTL_MOD);
        }
        else
        {
            _channels.insert(make_pair(channel->Fd(), channel));
            Updata(channel, EPOLL_CTL_ADD);
        }
    }
    // 移除Epoll中socket的监听
    void RemoveEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false)
            return;
        _channels.erase(channel->Fd());
        Updata(channel, EPOLL_CTL_DEL);
    }
    // 启动监听
    void Poll(vector<Channel *> &activate)
    {
        int num = epoll_wait(_epfd, _events, EPOLL_EVENT_MAX_SIZE, -1);
        if (num < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return;
            abort();
        }
        for (int i = 0; i < num; ++i)
        {
            struct epoll_event event = _events[i];
            int fd = event.data.fd;
            Channel *channel = _channels[fd];
            channel->SetREvent(event.events);
            activate.push_back(channel);
        }
    }
};
class TimeTask
{
    using CbType = std::function<void()>;
    using ReleaseFunc = std::function<void()>;

private:
    uint64_t _id;         // 超时任务ID
    uint32_t _timeout;    // 超时时间
    CbType _cb;           // 超时回调函数
    ReleaseFunc _release; // 删除回调函数
    bool _flag;           // 取消任务标志
public:
    TimeTask(uint64_t id, uint32_t time, const CbType &cb) : _id(id), _timeout(time), _cb(cb), _flag(false) {}
    ~TimeTask()
    {
        if (_flag == false)
            _cb();
        _release();
    }
    // 获取任务触发时间
    uint32_t GetTime()
    {
        return _timeout;
    }
    // 设置删除函数
    void SetRelease(const ReleaseFunc &cb)
    {
        _release = cb;
    }
    // 设置取消标志
    void CancleTask()
    {
        _flag = true;
    }
};
class TimeWheel
{
    using CbType = std::function<void()>;
    using TaskPtr = shared_ptr<TimeTask>;
    using TaskWeak = weak_ptr<TimeTask>;

private:
    int _trik;                                // 当前所处下标
    int _capacity;                            // 时间轮容量
    vector<vector<TaskPtr>> _wheels;          // 时间轮
    unordered_map<uint64_t, TaskWeak> _timer; // 任务ID和对应实体的映射
    uint64_t _timerfd;                        // 时间间隔
    EventLoop *_loop;                         // 对应EventLoop对象
    unique_ptr<Channel> _channel;             // 对应文件描述符的channel管理

private:
    // 从哈希表中删除
    void Release(uint64_t id)
    {
        auto it = _timer.find(id);
        if (it == _timer.end())
            return;
        _timer.erase(id);
    }
    // 创建timerd
    static int CreateTimerFd()
    {
        int _timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (_timerfd == -1)
        {
            ERR_LOG("Create timerfd failed");
            abort();
        }
        itimerspec time;
        time.it_value.tv_sec = 1;
        time.it_value.tv_nsec = 0;
        time.it_interval.tv_sec = 1;
        time.it_interval.tv_nsec = 0;
        timerfd_settime(_timerfd, 0, &time, nullptr);
        return _timerfd;
    }
    // 读取timerfd
    int ReadTimerFd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if (ret < 0)
        {
            ERR_LOG("Read timerfd failed");
            abort();
        }
        return times;
    }
    // 时间轮运行
    void Trik()
    {
        _trik = (_trik + 1) % _capacity;
        _wheels[_trik].clear();
    }
    // 添加任务
    void AddTaskInLoop(uint64_t id, uint32_t timeout, const CbType &cb)
    {
        auto it = _timer.find(id);
        if (it != _timer.end())
            return;
        int idx = (_trik + timeout) % _capacity;
        TaskPtr task(new TimeTask(id, timeout, cb));
        _wheels[idx].push_back(task);
        _timer[id] = TaskWeak(task);
        task->SetRelease(bind(&TimeWheel::Release, this, id));
    }
    // 刷新任务
    void RefreshTimeInLoop(uint64_t id)
    {
        auto it = _timer.find(id);
        if (it == _timer.end())
            return;
        TaskPtr task = it->second.lock();
        int time = task->GetTime();
        _wheels[(_trik + time) % _capacity].push_back(task);
    }
    // 取消任务
    void CancelTimeInLoop(uint64_t id)
    {
        auto it = _timer.find(id);
        if (it == _timer.end())
            return;
        TaskPtr pt = it->second.lock();
        if (pt)
            pt->CancleTask();
    }
    void Ontime()
    {
        int count = ReadTimerFd();
        for (int i = 0; i < count; ++i)
            Trik();
    }

public:
    TimeWheel(EventLoop *loop) : _trik(0), _capacity(60), _wheels(_capacity),
                                 _timerfd(CreateTimerFd()), _loop(loop), _channel(new Channel(_timerfd, _loop))
    {
        _channel->SetReadCallBack(bind(&TimeWheel::Ontime, this));
        _channel->EnableRead();
    }
    // 通过loop执行添加任务
    void AddTask(uint64_t id, uint32_t timeout, const CbType &cb);
    // 通过loop执行刷新任务
    void RefreshTime(uint64_t id);
    // 通过loop执行取消任务
    void CancelTime(uint64_t id);
    // 是否存在延迟任务
    bool HasTimer(uint64_t id)
    {
        auto it = _timer.find(id);
        if (it == _timer.end())
            return false;
        return true;
    }
};
class EventLoop
{
private:
    using Functor = function<void()>;
    thread::id _threadId;         // OneLoopOneThread
    int _eventFd;                 // 防止阻塞在Epoll模型
    unique_ptr<Channel> _channel; //_eventFd对应的Channel对象
    Poller _poller;               // Epoll模型
    vector<Functor> _tasks;       // 任务队列
    mutex _mutex;
    TimeWheel _wheel;

private:
    // 运行所有任务队列中的任务
    void RunAllTask()
    {
        vector<Functor> tasks;
        {
            unique_lock<mutex> _lock(_mutex);
            tasks.swap(_tasks);
        }
        for (auto &cb : tasks)
            cb();
    }
    // 创建eventfd
    static int CreateEventFd()
    {
        int fd = eventfd(0, EFD_CLOEXEC || EFD_NONBLOCK);
        if (fd == -1)
        {
            ERR_LOG("Create EventFd failed");
            abort();
        }
        return fd;
    }
    // eventfd的读事件
    void ReadEventFd()
    {
        uint64_t val;
        int size = read(_eventFd, (void *)&val, sizeof(val));
        if (size < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("Read EventFd failed");
            abort();
        }
    }
    // eventfd的写事件
    void WriteEventFd()
    {
        uint64_t val;
        int size = write(_eventFd, (void *)&val, sizeof(val));
        if (size < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("Write EventFd failed");
            abort();
        }
    }

public:
    EventLoop()
        : _threadId(this_thread::get_id()), _eventFd(CreateEventFd()), _channel(new Channel(_eventFd, this)), _wheel(this)
    {
        _channel->SetReadCallBack(bind(&EventLoop::ReadEventFd, this));
        _channel->EnableRead();
    }
    void AseertInLoop()
    {
        assert(_threadId == this_thread::get_id());
    }
    bool IsInLoop()
    {
        return (_threadId == this_thread::get_id());
    }
    // 执行传入函数
    void RunInLoop(const Functor &func)
    {
        if (IsInLoop())
            func();
        else
            QueueInLoop(func);
    }
    // 向任务队列中插入任务
    void QueueInLoop(const Functor &func)
    {
        {
            unique_lock<mutex> _lock(mutex);
            _tasks.push_back(func);
        }
        WriteEventFd(); // 防止阻塞在Epoll监听
    }
    // 插入或更新socket监听事件
    void UpdataEvent(Channel *channel) { _poller.UpdataEvent(channel); }
    // 删除socket监听事件
    void RemoveEvent(Channel *channel) { _poller.RemoveEvent(channel); }
    // 开始监听执行
    void Start()
    {
        while (1)
        {
            vector<Channel *> activates;
            _poller.Poll(activates);
            for (auto &e : activates)
                e->HandleEvent();
            RunAllTask();
        }
    }
    // 添加定时任务
    void AddTimer(uint64_t id, uint32_t time, const Functor &func)
    {
        _wheel.AddTask(id, time, func);
    }
    // 刷新定时任务
    void RefreshTimer(uint64_t id)
    {
        _wheel.RefreshTime(id);
    }
    // 取消定时任务
    void CancleTimer(uint64_t id)
    {
        _wheel.CancelTime(id);
    }
    // 是否存在定时任务
    bool HasTimer(uint64_t id)
    {
        return _wheel.HasTimer(id);
    }
};

class Acceptor
{
private:
    Socket _socket;   // 创建监听套接字
    EventLoop *_loop; // 对监听套接字进行事件监控
    Channel _channel; // 用于对监听套接字进行事件管理
    using AcceptCallback = function<void(int)>;
    AcceptCallback _accept_callback; // 回调处理函数
private:
    // 监听套接字并调用读回调函数
    void HandleRead()
    {
        int ret = _socket.Appect();
        if (ret < 0)
            return;
        // 将新连接以参数形式传递给业务处理函数
        if (_accept_callback)
            _accept_callback(ret);
    }
    // 创建监听套接字
    int CreateServer(int port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }

public:
    // 创建监听套接字并设置服务器回调函数
    Acceptor(EventLoop *loop, int port) : _socket(CreateServer(port)), _loop(loop), _channel(_socket.Fd(), loop)
    {
        _channel.SetReadCallBack(bind(&Acceptor::HandleRead, this));
    }
    // 设置回调函数
    void SetAcceptCallback(const AcceptCallback &cb) { _accept_callback = cb; }
    // 开启监听
    void Listen()
    {
        _channel.EnableRead();
    }
};

class Any
{
private:
    class placeholder
    {
    public:
        virtual ~placeholder() {}
        virtual const type_info &type() = 0;
        virtual placeholder *clone() = 0;
    };
    template <class T>
    class holder : public placeholder
    {
    public:
        holder(const T &val) : _val(val)
        {
        }
        holder(const holder &other) : _val(other._val)
        {
        }
        ~holder()
        {
        }
        const type_info &type()
        {
            return typeid(T);
        }
        placeholder *clone()
        {
            return new holder(_val);
        }
        T &getVal()
        {
            return _val;
        }

    private:
        T _val;
    };
    placeholder *_content;
    Any &swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

public:
    Any() : _content(nullptr) {}
    ~Any()
    {
        if (_content)
            delete _content;
    }
    template <class T>
    Any(const T &val) : _content(new holder<T>(val)) {}
    Any(const Any &other)
    {
        _content = other._content != nullptr ? other._content->clone() : nullptr;
    }
    template <class T>
    Any &operator=(const T &val)
    {
        Any tmp(val);
        swap(tmp);
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any tmp(other);
        swap(tmp);
        return *this;
    }
    template <class T>
    T &getVal()
    {
        assert(typeid(T) == _content->type());
        return ((holder<T> *)_content)->getVal();
    }
};

class Connection;
using ConnectionPtr = shared_ptr<Connection>;
enum ConnStatu
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
};
class Connection : public enable_shared_from_this<Connection>
{
    using ConnectedCallback = function<void(const ConnectionPtr &)>;
    using MessageCallback = function<void(const ConnectionPtr &, Buffer &)>;
    using ClosedCallback = function<void(const ConnectionPtr &)>;
    using AnyEventCallback = function<void(const ConnectionPtr &)>;

private:
    uint64_t _connId;
    int _sockfd;
    bool _flag;
    EventLoop *_loop;
    ConnStatu _statu;
    Socket _socket;
    Channel _channel;
    Buffer _inBuffer;
    Buffer _outBuffer;
    Any _context;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    ClosedCallback _server_closed_callback;
    AnyEventCallback _event_callback;

private:
    // 处理读事件
    void HandleRead()
    {
        char buffer[65536];
        int ret = _socket.NoBolckRecv(buffer, sizeof(buffer));
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return;
            return Shutdown();
        }
        _inBuffer.WriteAndPush(buffer, ret);
        if (_inBuffer.getReadSpace() > 0)
            _message_callback(shared_from_this(), _inBuffer);
    }
    // 处理写事件
    void HandleWrite()
    {
        int ret = _socket.NoBolckSend(_outBuffer.getReadAddr(), _outBuffer.getReadSpace());
        if (ret < 0)
        {
            if (_inBuffer.getReadSpace() > 0)
                _message_callback(shared_from_this(), _inBuffer);
            return Release();
        }
        _outBuffer.moveReadPosition(ret);
        if (_outBuffer.getReadSpace() == 0)
        {
            _channel.DisableWrite();
            if (_statu == DISCONNECTING)
                return Release();
        }
    }
    // 处理关闭事件
    void HandleClose()
    {
        if (_inBuffer.getReadSpace() > 0)
            _message_callback(shared_from_this(), _inBuffer);
        return Release();
    }
    // 处理错误事件
    void HandleErrno()
    {
        return HandleClose();
    }
    // 处理任意事件
    void HandleEvent()
    {
        if (_loop->HasTimer(_connId) == true)
            _loop->RefreshTimer(_connId);
        if (_event_callback)
            _event_callback(shared_from_this());
    }
    // 建立连接后的处理（监听读事件）
    void EstablishedInLoop()
    {
        assert(_statu == CONNECTING);
        _statu = CONNECTED;
        _channel.EnableRead();
        if (_connected_callback)
            _connected_callback(shared_from_this());
    }
    // 半关闭连接
    void ShutdownInLoop()
    {
        _statu = DISCONNECTING;
        if (_inBuffer.getReadSpace() > 0)
            _message_callback(shared_from_this(), _inBuffer);
        if (_outBuffer.getReadSpace() > 0)
        {
            if (_channel.WriteAble() == false)
                _channel.EnableWrite();
        }
        if (_outBuffer.getReadSpace() == 0)
            return Release();
    }
    // 释放连接
    void ReleaseInLoop()
    {
        _statu = DISCONNECTED;
        _channel.Remove();
        _socket.Close();
        if (_loop->HasTimer(_connId))
            CancleInactivateReleaseInLoop();
        if (_closed_callback)
            _closed_callback(shared_from_this());
        if (_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }
    // 想写缓冲区内写入数据
    void SendInLoop(Buffer buffer)
    {
        if (_statu == DISCONNECTED)
            return;
        _outBuffer.WriteBufferAndPush(buffer);
        if (_channel.WriteAble() == false)
            _channel.EnableWrite();
    }

    // 启动非活跃连接释放
    void EnableInactiveReleaseInLoop(int sec)
    {
        _flag = true;
        if (_loop->HasTimer(_connId))
            return _loop->RefreshTimer(_connId);
        _loop->AddTimer(_connId, sec, bind(&Connection::HandleClose, this));
    }
    // 关闭非活跃连接释放
    void CancleInactivateReleaseInLoop()
    {
        _flag = false;
        if (_loop->HasTimer(_connId))
            return _loop->CancleTimer(_connId);
    }
    // 更新协议上下文
    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn, const MessageCallback &mess,
                       const AnyEventCallback &event, const ClosedCallback &close)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = mess;
        _event_callback = event;
        _closed_callback = close;
    }

public:
    Connection(uint64_t connId, int sockfd, EventLoop *loop)
        : _connId(connId), _sockfd(sockfd), _flag(false), _statu(CONNECTING),
          _socket(_sockfd), _loop(loop), _channel(_sockfd, _loop)
    {
        _channel.SetWriteCallBack(bind(&Connection::HandleWrite, this));
        _channel.SetReadCallBack(bind(&Connection::HandleRead, this));
        _channel.SetCloseCallBack(bind(&Connection::HandleClose, this));
        _channel.SetErrnoCallBack(bind(&Connection::HandleErrno, this));
        _channel.SetEventCallBack(bind(&Connection::HandleEvent, this));
    }
    ~Connection()
    {
        DBG_LOG("~Connection : %p", get());
    }
    Connection *get()
    {
        return this;
    }
    // 获取管理的文件描述符去
    int Fd()
    {
        return _sockfd;
    }
    // 获取连接ID
    int Id()
    {
        return _connId;
    }
    // 是否处于CONNECTED状态
    bool Connected()
    {
        return _statu == CONNECTED;
    }
    // 设置上下文
    void SetContext(const Any &context)
    {
        _context = context;
    }
    // 获取上下文
    Any *GetContext()
    {
        return &_context;
    }
    // 设置回调函数
    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _event_callback = cb;
    }
    void SetSrvClosedCallback(const ClosedCallback &cb)
    {
        _server_closed_callback = cb;
    }
    // 建立连接就绪后，设置Channel回调
    void Established()
    {
        return _loop->RunInLoop(bind(&Connection::EstablishedInLoop, this));
    }
    // 发送数据
    void Send(const void *data, size_t len)
    {
        Buffer buffer;
        buffer.WriteAndPush(data, len);
        return _loop->RunInLoop(bind(&Connection::SendInLoop, this, move(buffer)));
    }
    // 关闭连接
    void Shutdown()
    {
        return _loop->RunInLoop(bind(&Connection::ShutdownInLoop, this));
    }
    void Release()
    {
        return _loop->RunInLoop(bind(&Connection::ReleaseInLoop, this));
    }
    // 启动非活跃监听
    void EnableInactiveRelease(int sec)
    {
        return _loop->RunInLoop(bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    // 关闭非活跃监听
    void CancleInactiveRelease()
    {
        return _loop->RunInLoop(bind(&Connection::CancleInactivateReleaseInLoop, this));
    }
    // 更新上下文
    void UpGrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &mess,
                 const AnyEventCallback &event, const ClosedCallback &close)
    {
        _loop->AseertInLoop();
        return _loop->RunInLoop(bind(&Connection::UpgradeInLoop, this, context, conn, mess, event, close));
    }
};

void Channel::Remove()
{
    _loop->RemoveEvent(this);
}
void Channel::Updata()
{
    _loop->UpdataEvent(this);
}
void TimeWheel::AddTask(uint64_t id, uint32_t timeout, const CbType &cb)
{
    _loop->RunInLoop(bind(&TimeWheel::AddTaskInLoop, this, id, timeout, cb));
}
void TimeWheel::RefreshTime(uint64_t id)
{
    _loop->RunInLoop(bind(&TimeWheel::RefreshTimeInLoop, this, id));
}
void TimeWheel::CancelTime(uint64_t id)
{
    _loop->RunInLoop(bind(&TimeWheel::CancelTimeInLoop, this, id));
}

class LoopThread
{
private:
    mutex _mutex;             // 互斥锁
    condition_variable _cond; // 条件变量
    EventLoop *_loop;         // 绑定的loop对象，必须在对应的线程中进行初始化
    thread _thread;           // 对应的线程
private:
    // 新建现成的入口函数 监听Epoll
    void ThreadEntry()
    {
        EventLoop loop;
        {
            unique_lock<mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        _loop->Start();
    }

public:
    LoopThread() : _loop(nullptr), _thread(&LoopThread::ThreadEntry, this) {}
    // 获取对象内的EventLoop
    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            unique_lock<mutex> lock(_mutex);
            _cond.wait(lock, [&]()
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    int _thread_count; // 从属EventLoop个数
    int _next_idx;
    EventLoop *_baseloop;          // 主EventLoop
    vector<EventLoop *> _loops;    // 管理从属EventLoop
    vector<LoopThread *> _threads; // 对应的从属LoopThread
public:
    LoopThreadPool(EventLoop *baseloop) : _thread_count(0), _next_idx(0), _baseloop(baseloop) {}
    // 设置从属EventLoop的个数
    void SetThreadCount(int count)
    {
        _thread_count = count;
    }
    // 创建多个从属EventLoop
    void Create()
    {
        if (_thread_count > 0)
        {
            _loops.resize(_thread_count);
            _threads.resize(_thread_count);
            for (int i = 0; i < _thread_count; ++i)
            {
                _threads[i] = new LoopThread;
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    // 获取EventLoop
    EventLoop *GetNextLoop()
    {
        if (_thread_count == 0)
            return _baseloop;
        _next_idx = (_next_idx + 1) % _thread_count;
        return _threads[_next_idx]->GetLoop();
    }
};
class TcpServer
{
private:
    uint64_t _next_id;                             // 分配connectID和延迟任务ID
    int _port;                                     // 监听端口
    int _timeout;                                  // 非活跃清除时间
    bool _flag;                                    // 是否启动非活跃清除
    EventLoop _baseloop;                           // 基础EventLoop 用于监听新连接
    Acceptor _acceptor;                            // 监听新连接
    LoopThreadPool _pool;                          // 从属EventLoop管理
    unordered_map<uint64_t, ConnectionPtr> _conns; // connectID与其本身的映射

    using ConnectedCallback = function<void(const ConnectionPtr &)>;
    using MessageCallback = function<void(const ConnectionPtr &, Buffer &)>;
    using ClosedCallback = function<void(const ConnectionPtr &)>;
    using AnyEventCallback = function<void(const ConnectionPtr &)>;
    using Functor = function<void()>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    ClosedCallback _server_closed_callback;
    AnyEventCallback _event_callback;

private:
    // 添加延迟任务
    void RunAfterTaskInLoop(const Functor &task, int time)
    {
        ++_next_id;
        _baseloop.AddTimer(_next_id, time, task);
    }
    // 对新建连接的处理
    void NewConnection(int fd)
    {
        ++_next_id;
        EventLoop *loop = _pool.GetNextLoop();
        ConnectionPtr ptr(new Connection(_next_id, fd, loop));
        ptr->SetConnectedCallback(_connected_callback);
        ptr->SetMessageCallback(_message_callback);
        ptr->SetClosedCallback(_closed_callback);
        ptr->SetAnyEventCallback(_event_callback);
        ptr->SetSrvClosedCallback(bind(&TcpServer::RemoveConnection, this, placeholders::_1));
        if (_flag)
            ptr->EnableInactiveRelease(_timeout);
        ptr->Established();
        _conns.insert(make_pair(_next_id, ptr));
    }
    // 移除映射
    void RemoveConnectionInLoop(const ConnectionPtr &conn)
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if (it != _conns.end())
            _conns.erase(it);
    }
    // 移除映射
    void RemoveConnection(const ConnectionPtr &conn)
    {
        _baseloop.RunInLoop(bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }

public:
    // 构造函数 启动Acceptor监听
    TcpServer(int port) : _next_id(0), _port(port), _timeout(0), _flag(false), _acceptor(&_baseloop, _port), _pool(&_baseloop)
    {
        _acceptor.SetAcceptCallback(bind(&TcpServer::NewConnection, this, placeholders::_1));
        _acceptor.Listen();
    }
    // 设置从属EventLoop个数
    void SetThreadCount(int count)
    {
        _pool.SetThreadCount(count);
    }
    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _closed_callback = cb;
    }
    void SetSrvClosedCallback(const ClosedCallback &cb)
    {
        _server_closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _event_callback = cb;
    }
    // 启动非活跃连接清理
    void EnableInactiveRelease(int time)
    {
        _timeout = time;
        _flag = true;
    }
    void CancleInactiveRelease()
    {
        _flag = false;
    }
    // 执行延迟任务
    void RunAfterTask(const Functor &task, int time)
    {
        _baseloop.RunInLoop(bind(&TcpServer::RunAfterTaskInLoop, this, task, time));
    }
    // 创建从属EventLoop并启动baseloop
    void Start()
    {
        _pool.Create();
        _baseloop.Start();
    }
};
class NetWork
{
public:
    NetWork()
    {
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;